package org.niit.service

import org.apache.spark.ml.recommendation.ALSModel
import org.apache.spark.sql.{DataFrame, SaveMode}
import org.apache.spark.streaming.dstream.DStream
import org.niit.bean.Answer
import org.niit.common.TService
import org.niit.util.SparkUtil

/**
 * Date:2025/6/11
 * Author：Ys
 * Description:
 */
class EDURecommendService {



   def dataAnalysis(data:DStream[Answer]): Unit ={
     val spakr = SparkUtil.takeSpark()
    import spakr.implicits._
    import org.apache.spark.sql.functions._

    data.foreachRDD(rdd=>{
      //1.声明模型的路径
      val path = "output/als_model/1749636734181"
      //2.加载模型
      val model = ALSModel.load(path)
      //3.自定义UDF方法：
      /*
         由于在训练模型的时候，不能有任务中文字样，所在在训练的时候将id进行的了切割
         所以在预测的时候，需要将id进行切割，才能进行预测
       */
      val id2Int = udf( (student_id:String)=>{
        student_id.split("_")(1).toInt
      })

      //4.由于ALS模型传入的参数必须DataFrame，所以要将rdd转换DataFrame
      val answerDF = rdd.toDF()

      //5.由于是针对学生Id进行推荐错题，所以要先将studentId查询并切割出来
      val studentIdDF = answerDF.withColumn("student_id", id2Int(col("student_id")))

      //6.使用模型给学生进行推荐题目 参数1：用户ID 参数2：推荐题目的数量
      val recommendDF: DataFrame = model.recommendForUserSubset(studentIdDF, 10)
      //打印推荐数据的数据结构
      recommendDF.printSchema()
      //打印推荐数据
      recommendDF.show(false)

      //7.处理推荐结果，取出用户id和题目id，将题目拼接成一个数组，将数组转换成字符串
      val res = recommendDF.as[ (Int,Array[ (Int,Float)]) ].map(t=>{
        val studentId = "学生ID_"+ t._1
        val questionIds = t._2.map(t=>"题目ID_"+t._1).mkString(",")
        (studentId,questionIds)
      }).toDF("student_id","recommendations")

      //8.将answerDF 和 res 进行 join连接，以为student_id进行关联
      val joinDF = answerDF.join(res, "student_id")

      //9.将joinDF的数据保存到数据库
      joinDF.write
        .format("jdbc")
        .option("url", "jdbc:mysql://node1:3306/BD1_2025?useUnicode=true&characterEncoding=utf8")
        .option("driver", "com.mysql.jdbc.Driver")
        .option("user", "root")
        .option("password", "Niit@123")
        .option("dbtable", "edu")
        .mode(SaveMode.Append)
        .save()
    })




  }

}
